package com.hrt.flink.icebergoperate.unauto;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink SQL API 批量/实时查询Iceberg表数据
 */
public class FlinkSQLIceberg2 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tblEnv = StreamTableEnvironment.create(env);

        Configuration configuration = tblEnv.getConfig().getConfiguration();
        // 支持SQL语法中的 OPTIONS 选项
        configuration.setBoolean("table.dynamic-table-options.enabled", true);

        env.enableCheckpointing(1000);

        //1.创建Catalog
        tblEnv.executeSql("create catalog hadoop_iceberg with ('type'='iceberg','catalog-type'='hadoop','warehouse'='hdfs://hadoop102:8020/flink_iceberg')");

        //2.批量读取Iceberg表数据
//        TableResult tableResult = tblEnv.executeSql("select * from hadoop_iceberg.iceberg_db.flink_iceberg_tbl2");
//        tableResult.print();

        //3.实时读取Iceberg表数据
//        TableResult tableResult = tblEnv.executeSql("SELECT * FROM hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s')*/");
//        tableResult.print();

        //4.基于某个快照id 实时增量读取数据
        TableResult tableResult = tblEnv.executeSql("SELECT * FROM hadoop_iceberg.iceberg_db.flink_iceberg_tbl2 /*+ OPTIONS('streaming'='true', 'monitor-interval'='1s', 'start-snapshot-id'='458759916745601336')*/");
        tableResult.print();



    }
}
